# -*- coding: utf-8 -*-
"""
--------------------------------------------
Date: 2024/4/17 9:18
Des: 能源数据
--------------------------------------------
"""
# 这段代码用于接收来自MQTT服务器的能源数据消息，解析消息内容，
# 并将解析后的数据分别写入TDengine和PostgreSQL数据库。它使用了线程池来异步写入数据库，以提高性能。
import json
import logging
import os
from concurrent.futures.thread import ThreadPoolExecutor
from paho.mqtt.client import Client, MQTTMessage
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'application.settings')
import django

django.setup()
from common.tdengine import load_tdengine_pool
from iot.models import Point
from django.db import close_old_connections

mqtt_host = '39.98.122.153'
mqtt_port = 1883
mqtt_username = 'emqx'
mqtt_password = 'public'
cache_point_type_cycle = 600
mqtt_client_id = 'sdasdjsaodjdasda'
point_type_map = {}
point_rule_map = {}
mqtt_topics = ['hanji/lcbgds/#']

logger = logging.getLogger(__name__)
thread_pool = ThreadPoolExecutor(max_workers=40)
tdengine_pool = load_tdengine_pool(object_pool_size=20)

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        logger.info(f'连接mqtt_broker{mqtt_host}:{mqtt_port}成功, 订阅主题列表: {mqtt_topics}')
        for topic in mqtt_topics:
            client.subscribe(topic, qos=0)
    else:
        logger.error(f'连接mqtt_broker{mqtt_host}:{mqtt_port}失败, 错误码为: {rc}')


def write_db(point_code: str, value, device_code: str, time_s):
    try:
        with tdengine_pool.get() as cursor:
            sql = f"INSERT INTO {point_code} USING iot TAGS ('{device_code}') VALUES ({time_s}, {value});"
            logger.info(f'Run Sql: {sql}')
            cursor.execute(sql)
    except Exception as e:
        logger.error(f'出错 point_code: {point_code}, value: {value}', exc_info=True)


def write_to_pg(point_code, value):
    points = Point.objects.filter(code=point_code)
    if not points:
        pass
    else:
        Point.update(point_code, **{'value': value})


# 接收消息回调
def on_message(client, userdata, msg: MQTTMessage):
    try:
        payload_data = json.loads(msg.payload.decode())
        data = payload_data.get('data', None)
        # 从解析后的JSON字典中尝试获取键为'data' 的值，如果不存在则返回None。
        data = payload_data.get('data', None)
        # 假设data是一个列表，并且列表的第一个元素是一个字典，从该字典中尝试获取键为'tp'的值，如果不存在则返回None。
        time_s = data[0].get('tp', None)
        # 同样地，从data列表的第一个字典元素中尝试获取键为'point'的值，如果不存在则返回None。
        point_data = data[0].get('point', None)
        if point_data:
            # 从 point_data 列表的第一个元素中获取键为 'val' 的值，并在其前面加上 'e_'，来构造一个 device_code 字符串。
            device_code = 'e_' + point_data[0].get('val')
            # 遍历 point_data 列表，从第二个元素开始到最后一个元素（即跳过第一个元素）。
            for each in point_data[1:]:
                # 调用 close_old_connections 函数，这个函数可能是用来关闭数据库连接的，以避免连接过多。
                close_old_connections()
                # 构造一个 point_code 字符串，它由 device_code 和当前元素的 'id' 键对应的值（转换为字符串）连接而成，中间用下划线分隔。
                point_code = device_code + '_' + str(each['id'])
                # 从当前元素中获取键为 'val' 的值，并将其赋值给变量 point_value。
                point_value = each['val']
                # 调用 write_to_pg 函数，将 point_code 和 point_value 作为参数传递，可能是将数据写入
                write_to_pg(point_code, point_value)
                # 使用 thread_pool 对象的 submit 方法异步执行 write_db 函数。write_db 可能是一个数据库写入函数，它接收 point_code、point_value、device_code 和 time_s 作为参数。
                thread_pool.submit(write_db, point_code, point_value, device_code, time_s)
    except Exception as e:
        logger.error(f'解析数据时出错: {msg.payload}', exc_info=True)


def run():
    client = Client(mqtt_client_id)
    client.username_pw_set(mqtt_username, mqtt_password)
    client.connect(host=mqtt_host, port=mqtt_port, keepalive=60)
    client.on_message = on_message
    client.on_connect = on_connect
    client.loop_forever()


if __name__ == '__main__':
    try:
        run()
    except Exception as e:
        logging.error('启动时意外退出', exc_info=True)
        os._exit(555)
